package com.houger.ddd.seckill.repository.mq;

import com.alibaba.fastjson.JSON;
import com.houger.ddd.common.utils.DateUtils;
import com.houger.ddd.seckill.facade.dto.mq.ActivityBasicMqDto;
import com.houger.ddd.seckill.facade.dto.mq.InstanceMqDto;
import com.houger.ddd.seckill.template.service.ActivityInstanceService;
import com.houger.ddd.seckill.template.service.ActivityTemplateService;
import com.rabbitmq.client.Channel;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;

/**
 * @author Houger.
 * @date 2024年05月28日 11:34.
 */
@Slf4j
@Component
public class RabbitMqComponent {
    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
    public static final String DELAYED_QUEUE_NAME = "delayed.queue";
    public static final String DELAYED_INSTANCE_QUEUE_NAME = "delayed.instance.queue";
    public static final String DELAYED_ROUTING_KEY = "delayed.routing.key";
    public static final String DELAYED_INSTANCE_ROUTING_KEY = "delayed.instance.routing.key";


    public static final String EXPIRE_EXCHANGE_NAME = "expire.exchange";
    public static final String EXPIRE_QUEUE_NAME = "expire.queue";
    public static final String EXPIRE_ROUTING_KEY = "expire.routing.key";


    public static final String DEAD_EXCHANGE_NAME = "dead.exchange";
    public static final String DEAD_QUEUE_NAME = "dead.queue";
    public static final String DEAD_ROUTING_KEY = "dead.routing.key";


    public static final String ORDER_EXCHANGE_NAME = "order.exchange";
    public static final String ORDER_QUEUE_NAME = "order.queue";
    public static final String ORDER_ROUTING_KEY = "order.routing.key";

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private ActivityTemplateService activityTemplateService;

    @Resource
    private ActivityInstanceService activityInstanceService;


    //===============(delay消息===============
    /**
     * @Retryable注解，并指定了需要重试的异常类型为AmqpException，
     * 最大重试次数为3次，重试间隔时间为5秒。
     * 当发送消息失败时，会触发重试逻辑。
     * 如果重试3次后仍然失败，则会执行recoverDelayedMessage方法中的恢复逻辑。
     * @date 2024/5/28 11:43
     * @param message
     * @param delay 设置延迟时间，单位为毫秒
     */
    @Retryable(retryFor = AmqpException.class, maxAttempts = 3, backoff = @Backoff(delay = 5000),recover = "recoversendDelayedMessage")
    public void sendDelayedMessage(String message, int delay) {
        log.info("delay={}",delay);
        rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message, message1 -> {
            message1.getMessageProperties().setDelay(delay);
            return message1;
        });
    }

    @Recover
    public void recoversendDelayedMessage(AmqpException e) {
        // 处理重试失败后的逻辑，比如记录日志、发送告警等
       log.error("Failed to send message after 3 attempts,e={}",e.getMessage());
    }

    /**
     * @Retryable注解表示该方法需要进行重试，其中
     * value属性指定了需要重试的异常类型，
     * maxAttempts属性指定了最大重试次数，
     * backoff属性指定了重试间隔时间。
     * 在retryMessage()方法中，使用rabbitTemplate发送消息到指定的交换器和路由键。
     *
     * x-delayed-type:	direct
     * @date 2024/5/29 15:50
     * @param messageData
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = DELAYED_QUEUE_NAME,durable = "true"),
        exchange = @Exchange(value = DELAYED_EXCHANGE_NAME),
        key = {DELAYED_ROUTING_KEY}
    ))
    public void handleDelayedMessage(String messageData, Message message, Channel channel) {
        try {
            log.info("handleDelayedMessage | [{}] messageData:[{}]",DateUtils.getNowFormat(),messageData);
            final ActivityBasicMqDto basicMqDto = JSON.parseObject(messageData, ActivityBasicMqDto.class);
            //活动状态变更：2未开始->3进行中
            final int i = activityTemplateService.updateStatus(basicMqDto.getTemplateId(), basicMqDto.getFromStatus(),
                basicMqDto.getToStatus(),basicMqDto.getVersion());
            if (i>0){
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                log.info("handleDelayedMessage | updateStatus res={}",i);
                return;
            }
            handleException(message,channel);
        }catch (Exception e){
            handleException(message,channel);
        }
    }

    @Retryable(retryFor = AmqpException.class, maxAttempts = 3, backoff = @Backoff(delay = 5000),recover = "recoversendDelayedMessageForInstance")
    public void sendDelayedMessageForInstance(String message, int delay) {
        log.info("sendDelayedMessageForInstance={}",delay);
        rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_INSTANCE_ROUTING_KEY, message, message1 -> {
            message1.getMessageProperties().setDelay(delay);
            return message1;
        });
    }

    @Recover
    public void recoversendDelayedMessageForInstance(AmqpException e) {
        // 处理重试失败后的逻辑，比如记录日志、发送告警等
        log.error("Failed to send message after 3 attempts,e={}",e.getMessage());
    }
    @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = DELAYED_INSTANCE_QUEUE_NAME,durable = "true"),
        exchange = @Exchange(value = DELAYED_EXCHANGE_NAME),
        key = {DELAYED_INSTANCE_ROUTING_KEY}
    ))
    public void handleDelayedMessageForInstance(String messageData, Message message, Channel channel) {
        try {
            log.info("handleDelayedMessageForInstance | [{}] messageData:[{}]",DateUtils.getNowFormat(),messageData);
            final InstanceMqDto instanceMqDto = JSON.parseObject(messageData, InstanceMqDto.class);
            //活动状态变更：2未开始->3进行中
            final int i = activityInstanceService.updateStatus(instanceMqDto.getInstanceId(), instanceMqDto.getFromStatus(),
                instanceMqDto.getToStatus(),instanceMqDto.getVersion());
            if (i>0){
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                log.info("handleDelayedMessageForInstance | updateStatus res={}",i);
                return;
            }
            handleException(message,channel);
        }catch (Exception e){
            handleException(message,channel);
        }
    }
    //===============delay消息)===============


    //===============(超时消息===============
    /**
     *
     * @date 2024/5/29 15:04
     * @param message
     * @param expire 设置过期时间，单位为毫秒
     */
    @Retryable(retryFor = AmqpException.class, maxAttempts = 3, backoff = @Backoff(delay = 5000),recover= "recoverSendExpirationMessage")
    public void sendExpirationMessage(String message, String expire) {
        rabbitTemplate.convertAndSend(EXPIRE_EXCHANGE_NAME, EXPIRE_ROUTING_KEY, message, message1 -> {
            message1.getMessageProperties().setExpiration(expire); // 设置过期时间，单位为毫秒
            return message1;
        });
    }
    @Recover
    public void recoverSendExpirationMessage(AmqpException e) {
        // 处理重试失败后的逻辑，比如记录日志、发送告警等
        log.error("Failed to send message after 3 attempts,e={}",e.getMessage());
    }

    @RabbitListener(queues = EXPIRE_QUEUE_NAME)
    public void handleExpirationMessage(String messageData, Message message, Channel channel) {
        try {
            log.info("handleExpirationMessage [{}] messageData:[{}]",DateUtils.getNowFormat(),messageData);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }catch (Exception e){
            handleException(message,channel);
        }
    }
    //===============超时消息)===============

    //===============（死信队列===============
    @Retryable(retryFor = AmqpException.class, maxAttempts = 3, backoff = @Backoff(delay = 5000),recover = "recoverSendOrderMessage")
    public void sendOrderMessage(String message, String expire) {
        rabbitTemplate.convertAndSend(ORDER_EXCHANGE_NAME, ORDER_ROUTING_KEY, message, message1 -> {
            if (StringUtils.isNotBlank(expire)){
                message1.getMessageProperties().setExpiration(expire); // 设置过期时间，单位为毫秒
            }
            return message1;
        });
    }
    @Recover
    public void recoverSendOrderMessage(AmqpException e) {
        // 处理重试失败后的逻辑，比如记录日志、发送告警等
        log.error("Failed to send message after 3 attempts,e={}",e.getMessage());
    }
//    @RabbitListener(queues = ORDER_QUEUE_NAME)
//    public void handleOrderMessage(String messageData, Message message, Channel channel) {
//        try {
//            log.info("handleOrderMessage [{}] messageData:{}",DateUtils.getNowFormat(),messageData);
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
//        }catch (Exception e){
//            handleException(message,channel);
//        }
//    }

    @RabbitListener(queues = DEAD_QUEUE_NAME)
    public void handleDeadMessage(String messageData, Message message, Channel channel) {
        try {
            log.info("handleDeadMessage [{}] messageData:[{}]", DateUtils.getNowFormat(),messageData);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }catch (Exception e){
            handleException(message,channel);
        }
    }


    //===============死信队列)===============
    private void handleException(Message message,Channel channel){
        final long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //            是否重复投递
            if (message.getMessageProperties().getRedelivered()){
                channel.basicNack(deliveryTag,false,true);
            }
            channel.basicNack(deliveryTag,false,false);
        } catch (IOException ex) {
            log.error(ex.getMessage());
        }
    }
    

}
